package com.pisolution.scheduler.runtime.master.consumer;

import com.pisolution.scheduler.core.tuple.Tuple3;
import com.pisolution.scheduler.core.utils.Stopper;
import com.pisolution.scheduler.datasource.instance.TaskInstance;
import com.pisolution.scheduler.runtime.context.TaskExecutionContext;
import com.pisolution.scheduler.runtime.context.TaskExecutionContextBuilder;
import com.pisolution.scheduler.runtime.master.MasterConfig;
import com.pisolution.scheduler.runtime.master.dispatcher.ExecutorDispatcher;
import com.pisolution.scheduler.runtime.process.ProcessService;
import com.pisolution.scheduler.runtime.queue.TaskQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 用于扫描 {@link TaskQueue} 并将扫描到的任务转交给工作节点执行.
 *
 * @author sora
 * @version 0.1.0
 */
@Component
public class TaskQueueConsumer extends Thread {
    private static final Logger LOGGER = LoggerFactory.getLogger(TaskQueueConsumer.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于从配置文件中读取一次性取出的队列中任务数量. */
    private final MasterConfig config;

    /** 任务存储队列. */
    private final TaskQueue<String> queue;

    /** 用于读取任务相关信息. */
    private final ProcessService service;

    /** 用于分发任务. */
    private final ExecutorDispatcher dispatcher;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化 {@link TaskQueueConsumer}.
     *
     * @param config 用于从配置文件中读取一次性取出的队列中任务数量.
     * @param queue 任务存储队列.
     * @param service 用于读取任务相关信息.
     * @param dispatcher 用于分发任务.
     */
    public TaskQueueConsumer(
        final MasterConfig config,
        final TaskQueue<String> queue,
        final ProcessService service,
        final ExecutorDispatcher dispatcher
    ) {
        this.config = config;
        this.queue = queue;
        this.service = service;
        this.dispatcher = dispatcher;
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 依赖注入完成后 启动扫描器扫描并处理任务.
     *
     * <p><b>注意:</b></p>
     * <p>该方法会在构造函数与依赖注入之后自动调用 无需手动触发.</p>
     */
    @SuppressWarnings("unused")
    @PostConstruct
    public void configure() {
        super.setName("task-queue-consumer");
        super.start();
    }

    // -------------------------------------------------------------------------------------------------
    //  任务处理
    // -------------------------------------------------------------------------------------------------

    @SuppressWarnings("BusyWait")
    @Override
    public void run() {
        // 1. 创建一个临时的存储介质存储处理失败的任务.
        final List<String> failedTasks = new ArrayList<>();
        final int fetchTaskNum = this.config.getMasterDispatchTaskNum();

        // 2. 如果管理节点尚未退出 则不断扫描任务队列.
        while (Stopper.isRunning()) {
            try {
                // 3. 清除上一次处理失败的任务.
                failedTasks.clear();

                // 4. 从任务队列中取出给定数量的任务.
                for (int count = 0; count < fetchTaskNum; count++) {
                    // 5. 如果任务队列为空 则等待可用任务.
                    if (this.queue.size() == 0) {
                        Thread.sleep(1_000L);

                        continue;
                    }

                    // 6. 取出任务并尝试发送任务.
                    final String taskInfo = this.queue.take();
                    if (this.dispatch(Integer.parseInt(taskInfo.split("_")[3]))) {
                        continue;
                    }

                    // 7. 将发送失败的任务追加到临时存储介质中.
                    failedTasks.add(taskInfo);
                }

                // 8. 如果没有发送失败的任务 则进入下一批次循环.
                if (failedTasks.isEmpty()) {
                    continue;
                }

                // 9. 将失败的任务逐条退回任务队列中.
                for (final String failedTask: failedTasks) {
                    this.queue.put(failedTask);
                }

                // 10. 如果任务出现循环执行 (即某些任务不断执行失败) 则等待一段时间后继续执行.
                if (this.queue.size() <= failedTasks.size()) {
                    TimeUnit.MILLISECONDS.sleep(1_000L);
                }
            } catch (Exception e) {
                LOGGER.error("Could not dispatch the task to the worker server.", e);
            }
        }
    }

    /**
     * 将给定任务分发给工作节点执行.
     *
     * @param taskId 待执行任务实例标识.
     *
     * @return 如果分发成功则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean dispatch(final int taskId) {
        try {
            // 1. 配置任务执行的上下文环境.
            final TaskInstance task = Objects.requireNonNull(this.service.findTaskInstanceById(taskId));

            task.setExecutePath("/home" + String.join(
                "/",
                String.valueOf(task.getUserId()),
                "scheduler",
                String.valueOf(task.getWorkflow().getProjectId()),
                String.valueOf(task.getWorkflow().getId()),
                String.valueOf(task.getWorkflowInstanceId()),
                String.valueOf(task.getId())
            ));

            final TaskExecutionContext context = new TaskExecutionContextBuilder()
                .buildWorkflow(task.getWorkflow())
                .buildWorkflowInstance(task.getWorkflowInstance())
                .buildTaskInstance(task)
                .create();

            // 2. 如果任务已经执行完成 则无需再次分发.
            if (task.getState().isFinished()) {
                return true;
            }

            return this.dispatcher.dispatch(new Tuple3<>(null, context.convert(), context.getWorkerGroup()));
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }

        return false;
    }
}
